package cn.baker.delayqueue.util;


import cn.baker.delayqueue.config.DelayQueueProperties;

import cn.baker.delayqueue.entity.DelayQueueMessage;
import cn.baker.delayqueue.mapper.DelayQueueMessageMapper;
import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Throwables;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.lang.NonNull;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * 分布式延时队列工具类
 *
 * @author yuanyu
 */
@Slf4j
public class RedisDelayQueueUtil {

    private final RedissonClient redissonClient;
    private final DelayQueueMessageMapper delayQueueMessageMapper;
    private final ObjectMapper objectMapper;
    private final DelayQueueProperties delayQueueProperties;

    public RedisDelayQueueUtil(RedissonClient redissonClient, DelayQueueMessageMapper delayQueueMessageMapper, ObjectMapper objectMapper, DelayQueueProperties delayQueueProperties) {
        this.redissonClient = redissonClient;
        this.delayQueueMessageMapper = delayQueueMessageMapper;
        this.objectMapper = objectMapper;
        this.delayQueueProperties = delayQueueProperties;
    }

    /**
     * 内部使用 添加延迟队列
     *
     * @param value     队列值
     * @param delay     延迟时间
     * @param timeUnit  时间单位
     * @param queueCode 队列键
     */
    public boolean addDelayQueue(boolean saveToDb, @NonNull Object value, @NonNull long delay, @NonNull TimeUnit timeUnit, @NonNull String queueCode) throws JsonProcessingException {
        if (StringUtils.isBlank(queueCode) || value == null) {
            return false;
        }

        log.debug("不考虑性能消耗，真正消费的时间点，time={}", System.currentTimeMillis() + timeUnit.toMillis(delay));
        // String saveValue = objectMapper.writeValueAsString(value);

        String saveValue;
        if (value instanceof String) {
            saveValue = (String) value;
        } else {
            saveValue = JSON.toJSONString(value);
        }
        if (delay <  0) {
            log.info("addDelayQueue 消息堆积,queueCode={},value={}", queueCode, saveValue);
            delay = 0;
        }

        // 时间大于Constant.ADD_QUEUE_MAX_TIME的数据存入mysql
        long delayInMs = timeUnit.toMillis(delay);
        if (saveToDb && delayInMs > delayQueueProperties.getAddQueueMaxTime()) {
            long timeout = System.currentTimeMillis() + delayInMs;
            DelayQueueMessage delayQueueMessage = DelayQueueMessage
                    .builder()
                    .delayInMillis(timeout)
                    .payload(saveValue)
                    .queueCode(queueCode)
                    .build();
            delayQueueMessageMapper.insert(delayQueueMessage);
            log.info("addDelayQueue 延迟时间过长，加入数据库暂存,delay={},queueCode={},value={}", timeUnit.toSeconds(delay) + "秒", queueCode, saveValue);
            return true;
        }
        try {
            RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);


            // 设置过期时间
            blockingDeque.expire(Duration.ofMillis(delayQueueProperties.getMillis()));
            delayedQueue.expire(Duration.ofMillis(delayQueueProperties.getMillis()));


            int size = delayedQueue.size();
            if (size > delayQueueProperties.getMaxQueueSize()) {
                log.error("addDelayQueue 队列堆积，丢掉消息，size={}", size);
            }
            delayedQueue.offer(saveValue, delay, timeUnit);
            log.info("addDelayQueue 添加延时队列成功，队列键：{}，队列值：{}，队列大小：{}，延迟时间：{}", queueCode, JSON.toJSONString(value), size, timeUnit.toSeconds(delay) + "秒");
        } catch (Exception e) {
            log.error("addDelayQueue 添加延时队列失败，e={}", Throwables.getStackTraceAsString(e));
            throw new RuntimeException("添加延时队列失败");
        }
        return true;
    }

    /**
     * 添加延迟队列
     *
     * @param value     队列值
     * @param delay     延迟时间
     * @param timeUnit  时间单位
     * @param queueCode 队列键
     */
    public boolean addDelayQueue(@NonNull Object value, @NonNull long delay, @NonNull TimeUnit timeUnit, @NonNull String queueCode) throws JsonProcessingException {
        return addDelayQueue(true, value, delay, timeUnit, queueCode);
    }

    /**
     * 获取延迟队列
     *
     * @param queueCode
     * @param <T>
     */
    public <T> T getDelayQueue(@NonNull String queueCode) throws InterruptedException {
        if (StringUtils.isBlank(queueCode)) {
            return null;
        }
        RBlockingDeque<T> blockingDeque = redissonClient.getBlockingDeque(queueCode);
        return blockingDeque.poll();
    }


}